/* Definitions for the stream type.

   This file is part of khipu.

   khipu is free software: you can redistribute it and/or modify
   it under the terms of the GNU Lesser General Public License as published by
   the Free Software Foundation; either version 3 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU Lesser General Public License for more details.

   You should have received a copy of the GNU Lesser General Public License
   along with this program.  If not, see <https://www.gnu.org/licenses/>.  */

#include <cstdlib>
#include <cstdio>
#include <climits>
#include "khipu.hpp"

KP_DECLS_BEGIN

static void
fini_stream (finobj *self)
{
  interpreter *interp = interpreter::self ();
  evh_safeguard eg { interp };
  deref (((stream *)self)->close (interp));
}

spos spos::decode (object obj)
{
  spos ret;
  object head;

  if (array_p (obj))
    {
      head = xaref (obj, 0);
      ret.state = xaref (obj, 1);
    }
  else
    head = obj;

  ret.offset = fixint_p (head) ? as_int (head) : as<int64_t> (head);
  return (ret);
}

result<object> spos::encode (interpreter *interp)
{
  object ret = intobj (interp, this->offset);

  if (this->state != UNBOUND)
    {
      ret = KP_TRY (alloc_array (interp, 2, ret));
      xaref(ret, 1) = this->state;
    }

  kp_return (ret);
}

result<bool> stream::xseek (interpreter *interp, const spos& npos, int whence)
{
  spos tmp = npos;
  bool ret = KP_TRY (this->ops->seek (interp, *this, tmp, whence));
  if (ret)
    this->pos = tmp;

  return (ret);
}

result<stream*> stream::make (interpreter *interp, int mode,
                              unsigned int bufsiz, const xops *ops,
                              void *cookie)
{
  auto eg = KP_TRY (evh_guard::make (interp));
  valref bvec = KP_TRY (alloc_bvector (interp,
                                       bufsiz << ((mode & STRM_BID) / STRM_BID)));
  valref ilock = KP_TRY ((mode & STRM_NOLOCK) ?
                         UNBOUND : alloc_lock (interp, true));

  stream *ret = alloch<stream> ();
  ret->ops = ops;
  ret->cookie = cookie;
  ret->bvec = *bvec;
  ret->rdbuf.init (as_bvector(*bvec)->data, bufsiz);
  ret->rdbuf.bump (bufsiz);

  char *wrbuf = ret->rdbuf.start + ((mode & STRM_BID) ? bufsiz : 0);
  ret->wrbuf.init (wrbuf, bufsiz);
  ret->ilock = *ilock;
  ret->io_flags = mode;

  ret->extra = UNBOUND;
  ret->fini = fini_stream;
  ret->vo_full |= FLAGS_FINALIZABLE;
  interp->alval = ret->as_obj ();
  gc_register (interp, ret);

  if (!(ret->io_flags & STRM_APP))
    ret->pos = spos (0);
  else
    {
      ret->io_flags &= ~STRM_OFFSET;
      KP_VTRY (ret->seek (interp, fixint (0), SEEK_END));
    }

  return (ret);
}

int stream::bufsize () const
{
  return (as_bvector(this->bvec)->nbytes);
}

result<bool> stream::write_full (interpreter *interp,
                                 const void *src, uint64_t bytes)
{
  while (bytes > 0)
    {
      auto tmp = KP_TRY (this->xwrite (interp, src, bytes));
      if (tmp <= 0)
        {
          this->io_flags |= STRM_ERR;
          return (false);
        }
      
      src = (const char *)src + tmp, bytes -= tmp;
    }

  return (true);
}

result<bool> stream::rflush (interpreter *interp)
{
  if (this->io_flags & STRM_OFFSET)
    // The current offset is always valid - Just seek from the beginning.
    return (this->xseek (interp, this->pos, SEEK_SET));
  
  // Get the current offset and seek from the beginning.
  spos npos (0);
  auto rv = KP_TRY (this->xseek (interp, npos, SEEK_CUR));
  if (!rv)
    return (false);

  rv = KP_TRY (this->xseek (interp, npos, SEEK_SET));
  if (!rv)
    return (false);
  
  this->pos = npos;
  this->reset_buff ();
  return (true);
}

result<bool> stream::flush (interpreter *interp)
{
  if (!(this->io_flags & STRM_LASTW))
    return (true);
  return (this->wflush (interp));
}

result<bool> stream::refill (interpreter *interp)
{
  if (this->io_flags & STRM_EOS)
    return (false);

  auto nr = KP_TRY (this->xread (interp, this->rdbuf.start, this->bufsize ()));

  if (nr <= 0)
    {
      this->io_flags |= nr == 0 ? STRM_EOS : STRM_ERR;
      this->rdbuf.set (this->rdbuf.end);
      return (false);
    }

  this->rdbuf.set (this->rdbuf.start);
  this->rdbuf.end = this->rdbuf.start + nr;
  return (true);
}

result<int64_t> stream::read (interpreter *interp, void *dst, uint64_t bytes)
{
  int64_t cnt, nbytes = (int64_t)bytes;
  char *base = (char *)dst;

  if (this->need_wflush ())
    KP_VTRY (this->wflush (interp));

  this->set_lastr ();
  /* We need to loop because the buffer may not be fully
   * filled on each iteration... */
  while (nbytes > (cnt = this->rdbuf.left ()))
    {
      memcpy (dst, this->rdbuf.curr, cnt);
      dst = (char *)dst + cnt, nbytes -= cnt;

      bool rv = KP_TRY (this->refill (interp));
      if (!rv)
        { // Partial result.
          this->add_pos ((char *)dst - base);
          return ((char *)dst - base);
        }
    }

  if (this->io_flags & STRM_UTF8)
    {
      int tmp = utf8min (this->rdbuf.curr, nbytes);
      bytes -= nbytes - tmp;
      nbytes = tmp;
    }

  memcpy (dst, this->rdbuf.curr, nbytes);
  this->rdbuf.bump (nbytes);
  this->add_pos (bytes);
  return ((int64_t)bytes);
}

result<int64_t> stream::write (interpreter *interp,
                               const void *src, uint64_t bytes)
{
  int64_t nbytes = (int64_t)bytes;

  if (this->need_rflush ())
    KP_VTRY (this->rflush (interp));

  this->set_lastw ();
  int64_t cnt = min ((int64_t)this->wrbuf.left (), nbytes);

  /* For UTF-8 streams, we have to make sure that we don't
   * write an incomplete character. */
  if (this->io_flags & STRM_UTF8)
    cnt = utf8min ((const char *)src, cnt);
  
  memcpy (this->wrbuf.curr, src, cnt);
  this->wrbuf.bump (cnt);
  src = (const char *)src + cnt;

  if ((nbytes -= cnt) > 0)
    { // There's still data to write.
      bool rv = KP_TRY (this->write_full (interp, this->wrbuf.start,
                                          this->wrbuf.used ()));
      if (!rv)
        return (-1);
      else if (nbytes > this->bufsize ())
        { // Write directly to handle.
          int64_t tmp = nbytes - (nbytes % this->bufsize ());

          if (this->io_flags & STRM_UTF8)
            for (; (((const unsigned char *)src)[tmp] & 0xc0) == 0x80; ++tmp) ;

          rv = KP_TRY (this->write_full (interp, src, tmp));
          if (!rv)
            return (-1);
          
          src = (const char *)src + tmp, nbytes -= tmp;
        }

      memcpy (this->wrbuf.start, src, nbytes);
      this->wrbuf.set (this->wrbuf.start + nbytes);
    }

  this->add_pos (bytes);
  return ((int64_t)bytes);
}

result<int> stream::putb (interpreter *interp, int byte)
{
  if (this->need_rflush ())
    {
      bool rv = KP_TRY (this->rflush (interp));
      if (!rv)
        return (-1);
    }
  else if (this->wrbuf.left () == 0)
    {
      bool rv = KP_TRY (this->wflush (interp));
      if (!rv)
        return (-1);
    }

  this->set_lastw ();
  *this->wrbuf.curr++ = (unsigned char)byte;
  this->add_pos (1);
  return (1);
}

result<int> stream::nputb (interpreter *interp, uint32_t n, int byte)
{
  if (n == 0)
    return (0);
  else if (this->need_rflush ())
    KP_VTRY (this->rflush (interp));

  int off = (int)n, cnt = min (this->wrbuf.left (), off);
  uint32_t bsz = this->bufsize ();

  memset (this->wrbuf.curr, byte, cnt);
  this->wrbuf.bump (cnt);
  this->set_lastw ();

  if ((n -= cnt) > 0)
    {
      bool rv = KP_TRY (this->write_full (interp, this->wrbuf.start, bsz));
      if (!rv)
        return (-1);

      for (cnt = n / bsz; cnt != 0; --cnt)
        {
          char *p = (char *)memset (this->wrbuf.start, byte, bsz);
          rv = KP_TRY (this->write_full (interp, p, bsz));
          if (!rv)
            return (-1);
        }
      
      cnt = n % bsz;
      memset (this->wrbuf.start, byte, cnt);
      this->wrbuf.set (this->wrbuf.start + cnt);
    }

  this->add_pos (off);
  return (off);
}

result<int> stream::putuc (interpreter *interp, uint32_t ch)
{
  unsigned char buf[4];
  int i = 0, len = u32tou8 (buf, ch);

  if (this->need_rflush ())
    KP_VTRY (this->rflush (interp));
  if (this->wrbuf.left () == 0)
    KP_VTRY (this->wflush (interp));

  this->set_lastw ();
  
  do
    *this->wrbuf.curr++ = buf[i++];
  while (i < len);

  this->add_pos (i);
  return (i);
}

result<int> stream::wbase (interpreter *interp, int base)
{
  char s[4];   // Enough for "36r
  char *p = s;
  
  switch (base)
    {
      case 10:
        return (0);
      case 8:
        *p++ = '0';
        break;
      case 16:
        p[0] = '0', p[1] = 'x';
        p += 2;
        break;
      case 2:
        p[0] = '0', p[1] = 'b';
        p += 2;
        break;
      default:
        {
          int c1 = base / 10;
          int c2 = base % 10;

          if (c1 != 0)
            *p++ = (char)(c1 + '0');
          *p++ = (char)(c2 + '0');
          *p++ = 'r';
          break;
        }
    }

  return (this->write (interp, s, (int)(p - s)));
}

result<int> stream::peekb (interpreter *interp)
{
  if (this->need_wflush ())
    KP_VTRY (this->wflush (interp));
  if (this->rdbuf.left () == 0)
    KP_VTRY (this->refill (interp));

  this->set_lastr ();
  return ((unsigned char)*this->rdbuf.curr);
}

result<int> stream::getb (interpreter *interp)
{
  int ret = KP_TRY (this->peekb (interp));
  this->rdbuf.bump (1);
  this->add_pos (1);
  return (ret);
}

static result<bool>
strm_sgetc (interpreter *interp, stream& strm, schar& ch)
{
  if (strm.need_wflush ())
    {
      bool rv = KP_TRY (strm.wflush (interp));
      if (!rv)
        return (rv);
    }

  if (strm.rdbuf.left () == 0)
    {
      bool rv = KP_TRY (strm.refill (interp));
      if (!rv)
        return (rv);
    }

  strm.set_lastr ();
  ch.buf[(ch.len = 1) - 1] = (unsigned char)*strm.rdbuf.curr++;
  int i = UTF8_SKIP[(uint8_t)ch.buf[0]];

  if (i == 0)
    return (false);
  for (int j = 1; j < i; ++j)
    {
      if (strm.rdbuf.left () == 0)
        {
          bool rv = KP_TRY (strm.refill (interp));
          if (!rv)
            return (false);
        }

      ch.buf[ch.len++] = (uint8_t)*strm.rdbuf.curr++;
    }

  ch.uc = u8tou32 (ch.buf, ch.len);
  strm.add_pos (ch.len);
  return (true);
}

result<bool> stream::sgetc (interpreter *interp, schar& ch)
{
  bool ret = KP_TRY (strm_sgetc (interp, *this, ch));
  if (!ret)
    ch.uc = UEOF;

  return (ret);
}

result<uint32_t> stream::getuc (interpreter *interp)
{
  schar ch;
  KP_VTRY (strm_sgetc (interp, *this, ch));
  return (ch.uc);
}

result<bool> stream::seek (interpreter *interp, object off, int whence)
{
  bool ret = false;

  if (this->ops->seek &&
      (whence == SEEK_SET || whence == SEEK_CUR || whence == SEEK_END))
    {
      if (this->need_wflush ())
        KP_VTRY (this->wflush (interp));

      ret = KP_TRY (this->xseek (interp, spos::decode (off), whence));
      if (ret)
        {
          this->reset_buff ();
          this->io_flags &= ~(STRM_EOS | STRM_ERR);
        }
    }

  return (ret);
}

bool stream::ungetuc (const unsigned char *p, int n)
{
  if (!(this->io_flags & (STRM_LASTR | STRM_BID)) ||
      n <= 0 || this->rdbuf.used () < n)
    return (false);

  this->add_pos (-n);

  do
    *--this->rdbuf.curr = p[--n];
  while (n != 0);

  return (true);
}

result<void> stream::lock (interpreter *interp)
{
#ifndef KP_NO_THREADS
  if (!(this->io_flags & STRM_NOLOCK))
    KP_VTRY (as_lock(this->ilock)->grab (interp, UNBOUND, false));
#endif
  return (0);
}

bool stream::trylock (interpreter *interp)
{
#ifndef KP_NO_THREADS
  return ((this->io_flags & STRM_NOLOCK) == 0 ||
           as_lock(this->ilock)->trygrab (interp) == 0);
#endif
}

void stream::unlock (interpreter *interp)
{
#ifndef KP_NO_THREADS
  if (!(this->io_flags & STRM_NOLOCK))
    as_lock(this->ilock)->drop (interp);
#endif
}

result<bool> stream::close (interpreter *interp)
{
  bool ret = true;
  
  if (this->io_flags & STRM_CLOSED)
    ;   // Nothing to do.
  else if (this->wrbuf.used () > 0)
    KP_VTRY (this->wflush (interp));
  
  if (this->ops->close &&
      (ret = this->ops->close (interp, *this)))
	this->io_flags |= STRM_CLOSED;

  return (ret);
}

static inline
fhandle_t as_osfh (void *cookie)
{
  return ((fhandle_t)(intptr_t)cookie);
}

static inline result<int64_t>
osfh_readb (interpreter *interp, stream& strm, void *dst, uint64_t nb)
{
  return (llio_readb (interp, as_osfh (strm.cookie), (char *)dst, nb));
}

static inline result<int64_t>
osfh_writeb (interpreter *interp, stream& strm, const void *ptr, uint64_t nb)
{
  return (llio_writeb (interp, as_osfh (strm.cookie), (const char *)ptr, nb));
}

static inline result<bool>
osfh_seekb (interpreter *interp, stream& strm, spos& pos, int whence)
{
  return (llio_seek (interp, as_osfh (strm.cookie), pos.offset, whence));
}

static inline bool
osfh_close (interpreter *, stream& strm)
{
  return (llio_close (as_osfh (strm.cookie)));
}

static const stream::xops bin_ops =
{
  osfh_readb,
  osfh_writeb,
  osfh_seekb,
  osfh_close
};

result<int64_t> llio_readt (interpreter *interp, fhandle_t fh,
                            char *dst, uint64_t nb)
{
  int64_t rv = KP_TRY (llio_readb (interp, fh, dst, nb - 1));
  if (rv <= 0)
    return (rv);

  char *inp = dst, *outp = dst;
  for (; inp != dst + rv - 1; ++inp)
    {
      int ch = *inp;
      if (ch == 13 && inp[1] == 10)
        {
          *outp++ = 10;
          ++inp;
        }
      else
        *outp++ = ch;
    }

  if (*inp == 13)
    { // Got a CR at the end - Peek for a LF.
      char b = 0;
      int64_t tmp = KP_TRY (llio_readb (interp, fh, &b, 1));

      if (tmp <= 0)
        *outp++ = 13;
      else if (b == 10)
        *outp++ = 10;
      else
        {
          *outp++ = 13;
          *outp++ = b;
        }
    }
  else
    *outp++ = *inp;

  return (outp - dst);
}

result<int64_t> llio_writet (interpreter *interp, fhandle_t fh,
                             const char *src, uint64_t nb)
{
  int64_t ret = 0;

  while (nb)
    {
      const uint32_t TSIZE = 1023;
      char buf[TSIZE + 1];
      const char *next = src, *src_end = src + nb, *lf;
      char *pbuf = buf, *pend = buf + TSIZE;
      uint64_t next_size = min ((uint64_t)TSIZE, nb);

      while (next_size > 0 &&
             (lf = (const char *)memchr (next, 10, next_size)))
        {
          uint32_t len = (uint32_t)(lf - next);
          memcpy (pbuf, next, len);
          pbuf[len] = 13, pbuf[len + 1] = 10;
          pbuf += len + 2;
          next += len + 1;
          next_size = min (src_end - next,
                           max ((ptrdiff_t)0, pend - pbuf));
        }

      if (next_size)
        {
          memcpy (pbuf, next, next_size);
          pbuf += next_size;
          next += next_size;
        }

      char *wbuf = buf;
      for (uint32_t nw = (uint32_t)(pbuf - buf) ; nw ; )
        {
          int64_t tmp = KP_TRY (llio_writeb (interp, fh, wbuf, nw));
          if (tmp <= 0)
            return (tmp);
          
          wbuf += tmp;
          nw -= tmp;
        }

      ret += next - src;
      src += ret;
      nb -= ret;
    }

  return (ret);
}

static inline result<int64_t>
osfh_readt (interpreter *interp, stream& strm, void *dst, uint64_t nb)
{
  return (llio_readt (interp, as_osfh (strm.cookie), (char *)dst, nb));
}

static inline result<int64_t>
osfh_writet (interpreter *interp, stream& strm, const void *ptr, uint64_t nb)
{
  return (llio_writet (interp, as_osfh (strm.cookie), (const char *)ptr, nb));
}

static inline result<bool>
osfh_seekt (interpreter *interp, stream& strm, spos& pos, int whence)
{
  if (pos.offset != 0)
    return (false);

  return (osfh_seekb (interp, strm, pos, whence));
}

static const stream::xops text_ops =
{
  osfh_readt,
  osfh_writet,
  osfh_seekt,
  osfh_close
};

result<stream*> fstream_open (interpreter *interp,
                              const char *path, const char *mode)
{
  int flags = 0;
  fhandle_t fh = llio_open (interp, path, mode, &flags);

  if (!fhandle_valid_p (fh))
    return (nullptr);

  const auto ops = (flags & STRM_BIN) ? &bin_ops : &text_ops;
  auto ret = stream::make (interp, flags, STRM_BUFSIZ, ops,
                           (void *)(intptr_t)fh);

  if (error_p (ret))
    {
      llio_close (fh);
      return (exception ());
    }
  else if ((*ret)->io_flags & STRM_WRITE)
    // Need to flush this object on exit.
    (*ret)->vo_full |= FLAGS_ATEXIT;

  return (*ret);
}

result<bool> fstream_fstat (interpreter *interp,
                            stream *strm, fhandle_stat& out)
{
  return (llio_fhstat (interp, as_osfh (strm->cookie), out));
}

result<bool> fstream_truncate (interpreter *interp,
                               stream *strm, uint64_t nsize)
{
  bool rv = KP_TRY (strm->flush (interp));
  return (rv && llio_truncate (as_osfh (strm->cookie), nsize));
}

// Standard streams initialisation.

struct std_stream
{
  bvector bvec;
  unsigned  char buffer[STRM_BUFSIZ];
  stream strm;
};

static result<stream*>
init_std_stream (interpreter *interp, std_stream& s,
                 uint32_t flags, fhandle_t fh)
{
  stream& strm = s.strm;

  strm.rdbuf.init (s.buffer, sizeof (s.buffer));
  strm.rdbuf.bump (sizeof (s.buffer));
  strm.wrbuf.init (s.buffer, sizeof (s.buffer));

  s.bvec.data = s.buffer;
  s.bvec.nbytes = sizeof (s.buffer);
  s.bvec.vo_type = typecode::BVECTOR;
  strm.bvec = s.bvec.as_obj ();

  strm.ops = &bin_ops;
  strm.ilock = KP_TRY (alloc_lock (interp, true));

  strm.vo_type = typecode::STREAM;
  strm.pos = spos (0);
  strm.io_flags = STRM_UTF8 | STRM_BUILTIN | flags;
  strm.cookie = (void *)(intptr_t)fh;
  strm.extra = UNBOUND;

  return (&strm);
}

// External definitions.
object in_stream;
object out_stream;
object err_stream;

static int
do_init_streams (interpreter *interp)
{
  int ret = init_op::call_deps (interp, &init_symbols, &init_llio);
  if (ret != init_op::result_ok)
    return (ret);

  static std_stream std_streams[3];

#define INIT_STREAM(ix, flags, rstream)   \
  do   \
    {   \
      auto p = init_std_stream (interp, std_streams[ix],   \
                                flags, llio_stdfh (ix));   \
      if (p.error_p ())   \
        return (init_op::result_failed);   \
      rstream = ensure_mask(*p)->as_obj ();   \
    }   \
  while (false)

  INIT_STREAM (0, STRM_READ, in_stream);
  INIT_STREAM (1, STRM_WRITE, out_stream);
  INIT_STREAM (2, STRM_WRITE, err_stream);

#undef INIT_STREAM

#define INTERN(name, out)   \
  do   \
    {   \
      auto tmp = intern (interp, name, sizeof (name) - 1,   \
                         nullptr, symbol::special_flag);   \
      if (tmp.error_p ())   \
        return (init_op::result_failed);   \
      symval(*tmp) = out;   \
    }   \
  while (false)

  // Add the dynamic bindings.
  INTERN ("*in*", in_stream);
  INTERN ("*out*", out_stream);
  INTERN ("*err*", err_stream);

#undef INTERN

  return (init_op::result_ok);
}

init_op init_streams (do_init_streams, "streams");

// XXX: Better definition.
const object EOS = fixint (0) | EXTRA_BIT;

KP_DECLS_END
